#include "discovery/eventfd_domain_socket_receiver.h"
#include "discovery/eventfd_types.h"
#include "log/logger.h"
#include <algorithm>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

EventFdDomainSocketReceiver::ConnectionInfo::ConnectionInfo()
    : socket_fd(-1), connect_time(std::chrono::steady_clock::now()) {}

void EventFdDomainSocketReceiver::ConnectionInfo::reset() {
  if (descriptor) {
    descriptor->close();
    descriptor.reset();
  }
  if (socket_fd >= 0) {
    close(socket_fd);
    socket_fd = -1;
  }
}

EventFdDomainSocketReceiver::EventFdDomainSocketReceiver(
    io_context &io, const std::string &socket_path,
    const std::string &identifier)
    : EventFdReceiver(io, identifier), socket_path_(socket_path),
      server_socket_(-1) {

  cleanup_timer_ = std::make_unique<steady_timer>(io_);
}

EventFdDomainSocketReceiver::~EventFdDomainSocketReceiver() { stop(); }

bool EventFdDomainSocketReceiver::start() {
  if (running_.exchange(true)) {
    return false;
  }

  try {
    return start_server();
  } catch (const std::exception &e) {
    JC_ERROR("启动接收器失败: {}", e.what());
    running_ = false;
    return false;
  }
}

void EventFdDomainSocketReceiver::stop() {
  if (!running_.exchange(false)) {
    return;
  }

  if (cleanup_timer_)
    cleanup_timer_->cancel();

  if (server_descriptor_) {
    server_descriptor_->close();
    server_descriptor_.reset();
  }

  if (server_socket_ >= 0) {
    close(server_socket_);
    server_socket_ = -1;
    unlink(socket_path_.c_str());
  }

  std::lock_guard<std::mutex> lock(connections_mutex_);
  connections_.clear();
}

size_t EventFdDomainSocketReceiver::connection_count() const {
  std::lock_guard<std::mutex> lock(connections_mutex_);
  return connections_.size();
}

void EventFdDomainSocketReceiver::set_extended_transfer_callback(ExtendedTransferCallback callback) {
  extended_transfer_callback_ = std::move(callback);
}

bool EventFdDomainSocketReceiver::start_server() {

  unlink(socket_path_.c_str());

  server_socket_ = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);
  if (server_socket_ < 0) {
    throw std::runtime_error("Failed to create server socket");
  }

  struct sockaddr_un addr;
  memset(&addr, 0, sizeof(addr));
  addr.sun_family = AF_UNIX;
  strncpy(addr.sun_path, socket_path_.c_str(), sizeof(addr.sun_path) - 1);

  if (bind(server_socket_, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
    close(server_socket_);
    throw std::runtime_error("Failed to bind server socket");
  }

  if (listen(server_socket_, 10) < 0) {
    close(server_socket_);
    throw std::runtime_error("Failed to listen on server socket");
  }

  server_descriptor_ =
      std::make_unique<posix::stream_descriptor>(io_, server_socket_);

  start_accept();

  start_cleanup_timer();

  JC_INFO("EventFD接收器启动，监听: {}", socket_path_);
  return true;
}

void EventFdDomainSocketReceiver::start_accept() {
  if (!running_)
    return;

  server_descriptor_->async_wait(
      posix::stream_descriptor::wait_read,
      [this](const boost::system::error_code &ec) {
        if (!ec) {
          handle_accept();
          start_accept();
        } else if (ec != boost::asio::error::operation_aborted) {
          JC_ERROR("接受连接错误: {}", ec.message());
          if (running_) {

            cleanup_timer_->expires_after(std::chrono::milliseconds(1000));
            cleanup_timer_->async_wait(
                [this](const boost::system::error_code &) { start_accept(); });
          }
        }
      });
}

void EventFdDomainSocketReceiver::handle_accept() {
  int client_socket = accept(server_socket_, nullptr, nullptr);
  if (client_socket < 0) {
    return;
  }

  int flags = fcntl(client_socket, F_GETFL, 0);
  fcntl(client_socket, F_SETFL, flags | O_NONBLOCK);

  auto connection = std::make_unique<ConnectionInfo>();
  connection->socket_fd = client_socket;
  connection->descriptor =
      std::make_unique<posix::stream_descriptor>(io_, client_socket);
  connection->peer_info = "client_" + std::to_string(client_socket);

  JC_INFO("新客户端连接: {}", connection->peer_info);

  start_receive_eventfd(connection.get());

  {
    std::lock_guard<std::mutex> lock(connections_mutex_);
    connections_.push_back(std::move(connection));
  }

  if (connection_callback_) {
    connection_callback_(boost::system::error_code{}, true);
  }
}

void EventFdDomainSocketReceiver::start_receive_eventfd(
    ConnectionInfo *connection) {
  if (!running_ || !connection || !connection->descriptor) {
    JC_DEBUG("receiver check failed: running={}, connection={}, descriptor={}",
             running_.load(),
             connection != nullptr,
             connection && connection->descriptor != nullptr);
    return;
  }

  // 直接等待接收EventFD，不再预先读取TransferMessage
  connection->descriptor->async_wait(
      posix::stream_descriptor::wait_read,
      [this, connection](const boost::system::error_code &ec) {
        if (!ec) {
          TransferMessage received_msg;
          receive_eventfd_from_connection(connection, received_msg);
        } else {
          JC_ERROR("等待接收EventFD失败: {}", ec.message());
          remove_connection(connection);
        }
      });
}

void EventFdDomainSocketReceiver::receive_eventfd_from_connection(
    ConnectionInfo *connection, TransferMessage& received_msg) {
  int received_fd = recv_fd_with_info(connection->socket_fd, received_msg);

  if (received_fd >= 0) {
    JC_INFO("从 {} 接收到EventFD: {} (类型: {}, 主题: {}, 数据: {})", 
            connection->peer_info, received_fd, 
            received_msg.get_message_type(), 
            received_msg.get_topic_name(), 
            received_msg.get_custom_data());

    // 优先使用扩展回调
    if (extended_transfer_callback_) {
      extended_transfer_callback_(boost::system::error_code{}, received_fd, received_msg);
    } else if (transfer_callback_) {
      transfer_callback_(boost::system::error_code{}, received_fd);
    }

    // 继续等待下一个消息
    start_receive_eventfd(connection);
  } else {
    JC_DEBUG("连接 {} 关闭或接收失败，移除连接", connection->peer_info);
    remove_connection(connection);
  }
}

void EventFdDomainSocketReceiver::start_cleanup_timer() {
  if (!running_)
    return;

  cleanup_timer_->expires_after(cleanup_interval_);
  cleanup_timer_->async_wait([this](const boost::system::error_code &ec) {
    if (!ec && running_) {
      cleanup_dead_connections();
      start_cleanup_timer();
    }
  });
}

void EventFdDomainSocketReceiver::cleanup_dead_connections() {
  std::lock_guard<std::mutex> lock(connections_mutex_);

  auto now = std::chrono::steady_clock::now();

  for (auto it = connections_.begin(); it != connections_.end();) {
    auto &connection = *it;

    auto duration = now - connection->connect_time;
    if (duration > std::chrono::minutes(5)) {
      JC_INFO("清理超时连接: {}", connection->peer_info);
      it = connections_.erase(it);
    } else {
      ++it;
    }
  }
}

void EventFdDomainSocketReceiver::remove_connection(ConnectionInfo *target) {
  std::lock_guard<std::mutex> lock(connections_mutex_);

  connections_.erase(
      std::remove_if(connections_.begin(), connections_.end(),
                     [target](const std::unique_ptr<ConnectionInfo> &conn) {
                       return conn.get() == target;
                     }),
      connections_.end());

  if (connection_callback_) {
    connection_callback_(
        make_error_code(boost::system::errc::connection_aborted), false);
  }
}

int EventFdDomainSocketReceiver::recv_fd(int socket_fd) {
  struct msghdr msg = {0};
  char m_buffer[256];
  char c_buffer[256];

  struct iovec io = {.iov_base = m_buffer, .iov_len = sizeof(m_buffer)};
  msg.msg_iov = &io;
  msg.msg_iovlen = 1;
  msg.msg_control = c_buffer;
  msg.msg_controllen = sizeof(c_buffer);

  if (recvmsg(socket_fd, &msg, 0) < 0) {
    return -1;
  }

  struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
  if (cmsg && cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
    return *((int *)CMSG_DATA(cmsg));
  }

  return -1;
}

int EventFdDomainSocketReceiver::recv_fd_with_info(int socket_fd, TransferMessage& msg) {
  struct msghdr msghdr_recv = {0};
  char c_buffer[256];

  // 使用TransferMessage作为接收缓冲区
  struct iovec io = {.iov_base = &msg, .iov_len = sizeof(msg)};
  msghdr_recv.msg_iov = &io;
  msghdr_recv.msg_iovlen = 1;
  msghdr_recv.msg_control = c_buffer;
  msghdr_recv.msg_controllen = sizeof(c_buffer);

  ssize_t bytes_received = recvmsg(socket_fd, &msghdr_recv, 0);
  if (bytes_received < 0) {
    JC_ERROR("recvmsg失败: {}", strerror(errno));
    return -1;
  }

  if (bytes_received == 0) {
    // 连接关闭，这是正常情况
    JC_DEBUG("连接已关闭");
    return -1;
  }

  if (bytes_received != sizeof(TransferMessage)) {
    JC_ERROR("接收到的消息大小不匹配: {} vs {}", bytes_received, sizeof(TransferMessage));
    return -1;
  }

  struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msghdr_recv);
  if (cmsg && cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
    int received_fd = *((int *)CMSG_DATA(cmsg));
    JC_DEBUG("成功接收EventFD: {} 和消息信息 (类型: {}, 主题: {})", 
             received_fd, msg.get_message_type(), msg.get_topic_name());
    return received_fd;
  }

  JC_ERROR("未找到文件描述符控制消息");
  return -1;
}
